package service

import (
	"gitee.com/sunny-zhangqiang/zcache_new/comm/fileutil"
	"gitee.com/sunny-zhangqiang/zcache_new/config"
	"gitee.com/sunny-zhangqiang/zcache_new/logger"
	"gitee.com/sunny-zhangqiang/zcache_new/ztp"
	uuid "github.com/satori/go.uuid"
	"net"
	"os"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
)

var (
	startTime     int64
	serverToken   string
	serviceQuit   int64
	serviceActive int64
	task          int64

	ztpServers []*ztp.ZTP
	srv        *server
)

type server struct {
	api    map[string]func(c *ztp.Conn, req *ztp.Request, resp *ztp.Response)
	client *ztp.ZTPClient
}

func (srv *server) Do(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) error {
	atomic.AddInt64(&task, 1)
	defer atomic.AddInt64(&task, -1)

	f := srv.api[req.Action]

	if f != nil {
		f(c, req, resp)
		return nil
	}

	return apiNotFoundErr
}

func (srv *server) get(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.get()
}

func (srv *server) put(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.put()
}

func (srv *server) del(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.del()
}

func (srv *server) getAndWait(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.getAndWait()
}

func (srv *server) putAndNotify(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.putAndNotify()
}

func (srv *server) revalidateWait(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.revalidateWait()
}

func (srv *server) revalidateNotify(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.revalidateNotify()
}

func (srv *server) putChunkFragment(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.putChunkFragment()
}

func (srv *server) getChunkFragment(c *ztp.Conn, req *ztp.Request, resp *ztp.Response) {
	r := request{}
	requestInit(&r, c, req, resp)
	r.getChunkFragment()
}

func newServer() *server {
	s := &server{}
	s.api = make(map[string]func(c *ztp.Conn, req *ztp.Request, resp *ztp.Response))
	s.api["/get"] = s.get
	s.api["/put"] = s.put
	s.api["/del"] = s.del
	s.api["/get_and_wait"] = s.getAndWait
	s.api["/put_and_notify"] = s.putAndNotify
	s.api["/revalidate_wait"] = s.revalidateWait
	s.api["/revalidate_notify"] = s.revalidateNotify
	s.api["/put_chunk_fragment"] = s.putChunkFragment
	s.api["/get_chunk_fragment"] = s.getChunkFragment

	s.client = ztp.NewZTPClient(128,
		10*1000, 3*1000,
		3*1000, 3*1000)
	return s
}

func startZTPServer() error {
	srv = newServer()
	ztpServers = make([]*ztp.ZTP, len(config.ServerConf.ZTPListeners))
	for i := 0; i < len(config.ServerConf.ZTPListeners); i++ {
		ztpServers[i] = ztp.NewZTP(srv)
		err := ztpServers[i].Start(config.ServerConf.ZTPListeners[i])
		if err != nil {
			return err
		}
	}
	return nil
}

func stopZTPServer() {
	if ztpServers == nil {
		return
	}

	wg := &sync.WaitGroup{}
	wg.Add(len(ztpServers))

	for _, s := range ztpServers {
		go func(zs *ztp.ZTP) {
			zs.Stop()
			wg.Done()
		}(s)
	}

	wg.Wait()
}

func startProbListen() error {
	l, err := net.Listen("tcp", config.ServerConf.ProbListen)
	if err != nil {
		return err
	}

	go func(ls net.Listener) {
		for {
			if atomic.LoadInt64(&serviceQuit) == 1 {
				break
			}

			c, err := ls.Accept()
			if err != nil {
				logger.Error("prob listen accept error: %s", err.Error())
				continue
			}

			go doProbProcess(c)
		}

		ls.Close()
	}(l)

	return nil
}

func genServerToken() string {
	u, err := uuid.NewV1()

	prefix := ""
	if err == nil {
		prefix = u.String()
	}

	return prefix + "#" +
		config.ServerConf.ServerName + "-" +
		strconv.Itoa(os.Getpid()) + "-" +
		strconv.FormatInt(startTime, 16)
}

func Start() error {
	startTime = time.Now().UnixMilli()
	serverToken = genServerToken()

	go func() {
		for {
			exists, _ := fileutil.FileExists(config.ServerConf.ActiveFile)
			if exists {
				atomic.StoreInt64(&serviceActive, 1)
			} else {
				atomic.StoreInt64(&serviceActive, 0)
			}
			time.Sleep(time.Second)
		}
	}()

	openMerge()
	openChunkMerge()

	err := startZTPServer()
	if err != nil {
		return err
	}

	err = startProbListen()
	if err != nil {
		return err
	}

	return nil
}

func NotifyStop() error {
	atomic.StoreInt64(&serviceQuit, 1)
	stopZTPServer()
	return nil
}

func WaitStop() error {
	for {
		if atomic.LoadInt64(&task) == 0 {
			return nil
		}

		time.Sleep(time.Millisecond * 100)
	}
}
